package io.kiki.sba.registry.server.data_store.remoting.sessionserver.handler;

import io.kiki.sba.registry.common.model.ClientInterfaceId;
import io.kiki.sba.registry.common.model.Node;
import io.kiki.sba.registry.common.model.PublishType;
import io.kiki.sba.registry.common.model.Response;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.common.model.slot.SlotAccess;
import io.kiki.sba.registry.common.model.slot.SlotAccessGenericResponse;
import io.kiki.sba.registry.common.model.store.ClientInterfaceIdIntern;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.common.model.store.StoreData;
import io.kiki.sba.registry.remoting.netty.Channel;
import io.kiki.sba.registry.server.data_store.bootstrap.ServerConfig;
import io.kiki.sba.registry.server.data_store.cache.IDatumStorage;
import io.kiki.sba.registry.server.data_store.change.DataChangeEventCenter;
import io.kiki.sba.registry.server.data_store.lease.ClientInterfaceLeaseManager;
import io.kiki.sba.registry.server.data_store.slot.SlotAccessorDelegate;
import io.kiki.sba.registry.server.shared.remoting.AbstractServerHandler;
import io.kiki.sba.registry.util.StringFormatter;
import io.kiki.sba.registry.util.StringUtil;
import lombok.Getter;
import lombok.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;


public abstract class AbstractDataHandler<T> extends AbstractServerHandler<T> {
    private static final Logger logger = LoggerFactory.getLogger(AbstractDataHandler.class);

    private final Set<StoreData.DataType> DATA_TYPES = Collections.unmodifiableSet(new HashSet<>(Arrays.asList(StoreData.DataType.publisher, StoreData.DataType.un_publisher)));

    @Getter
    @Autowired
    protected DataChangeEventCenter dataChangeEventCenter;

    @Autowired
    protected ServerConfig serverConfig;

    @Getter
    @Autowired
    protected SlotAccessorDelegate slotAccessorDelegate;

    @Getter
    @Resource
    protected IDatumStorage datumStorageDelegate;

    @Getter
    @Autowired
    protected ClientInterfaceLeaseManager clientInterfaceLeaseManager;

    protected void checkPublisher(@NonNull Publisher publisher) {

        StringUtil.checkNotBlank(publisher.getDataId(), "publisher.dataId");
        StringUtil.checkNotBlank(publisher.getInstanceId(), "publisher.instanceId");
        StringUtil.checkNotBlank(publisher.getGroup(), "publisher.group");
        StringUtil.checkNotBlank(publisher.getDataInfoId(), "publisher.dataInfoId");
        StringUtil.nullValue(publisher.getVersion(), "publisher.version");
        StringUtil.checkNotBlank(publisher.getRegisterId(), "publisher.registerId");
        if (publisher.getPublishType() != PublishType.temporary && publisher.getDataType() == StoreData.DataType.publisher) {
            StringUtil.nullValue(publisher.getClientUrl(), "publisher.sourceAddress");
        }
        if (!DATA_TYPES.contains(publisher.getDataType())) {
            throw new IllegalArgumentException(StringFormatter.format("{} is not contain in {}, {}", publisher.getDataType(), DATA_TYPES, "publisher.dataType"));
        }


    }

    protected void checkClientInterfaceId(@NonNull ClientInterfaceId clientInterfaceId) {

    }

    protected SlotAccess checkSlotAccess(String dataCenter, String dataInfoId, long slotsEpoch, long slotLeaderEpoch) {
        final int slotId = slotAccessorDelegate.slotOf(dataInfoId);
        return checkSlotAccess(dataCenter, slotId, slotsEpoch, slotLeaderEpoch);
    }

    protected SlotAccess checkSlotAccess(String dataCenter, int slotId, long slotsEpoch, long slotLeaderEpoch) {
        final SlotAccess slotAccess = slotAccessorDelegate.checkSlotAccess(dataCenter, slotId, slotsEpoch, slotLeaderEpoch);
        if (slotAccess.isMoved()) {
            logger.warn("[moved]{}, dataCenter={}, leaderEpoch={}, tableEpoch={}", slotAccess, dataCenter, slotLeaderEpoch, slotsEpoch);
        }

        if (slotAccess.isMigrating()) {
            logger.warn("[migrating]{}, dataCenter={}, leaderEpoch={}, tableEpoch={}", slotAccess, dataCenter, slotLeaderEpoch, slotsEpoch);
        }
        if (slotAccess.isMisMatch()) {
            logger.warn("[mismatch]{}, dataCenter={}, leaderEpoch={}, tableEpoch={}", slotAccess, dataCenter, slotLeaderEpoch, slotsEpoch);
        }
        return slotAccess;
    }

    protected void processClientInterfaceId(io.kiki.sba.registry.api.Channel channel, ClientInterfaceId clientInterfaceId) {
        // the channel is null when caller is xxx-Resource
        if (channel != null) {
            // bind the processId with the conn
            clientInterfaceId = ClientInterfaceIdIntern.intern(clientInterfaceId);
            final io.kiki.stack.netty.channelManager.Channel channel1 = ((Channel) channel).getChannel();
            channel1.setAttributeIfAbsent(ValueConstants.ttr_rpc_channel_client_interface_id, clientInterfaceId);
        }
        clientInterfaceLeaseManager.renewal(clientInterfaceId);
    }

    @Override
    public Response buildFailedResponse(String msg) {
        return SlotAccessGenericResponse.failedResponse(msg);
    }

    @Override
    protected Node.NodeType getConnectNodeType() {
        return Node.NodeType.client_interface;
    }

    @Override
    protected void logRequest(io.kiki.sba.registry.api.Channel channel, T request) {
        // too much to log
    }

}
